package db

import (
	"context"
	"fmt"
	"github.com/beego/beego/v2/client/orm"
	"scase.io/application-auto-scaling-service/pkg/common"
	"scase.io/application-auto-scaling-service/pkg/utils/logger"
	"sync"
	"time"
)

const (
	tableNamePodInfo    = "pod_info"
	PodStatePending     = "Pending" // pod创建中
	PodStateRunning     = "Running" // pod创建成功，容器运行中
	PodStateActive      = "Active"  // 容器中的应用运行成功
	PodStateDeleted     = "Deleted" // pod删除
	PodStateError       = "Error"
	PodStateDeleting    = "Deleting"
	fieldNameHostName   = "host_name"
	fieldNameHostNameIn = "host_name__in"
)

type PodInfo struct {
	HostName       string    `json:"host_name" orm:"column(host_name);size(128);pk"`
	ProjectId      string    `json:"project_id" orm:"column(project_id);size(128)"`
	FleetId        string    `json:"fleet_id" orm:"column(fleet_id);size(128)"`
	Namespace      string    `orm:"column(namespace);size(128)"`
	ScalingGroupId string    `json:"scaling_group_id" orm:"column(scaling_group_id);size(128)"`
	InstanceId     string    `json:"instance_id" orm:"column(instance_id);size(128)"`
	PublicIp       string    `json:"public_ip" orm:"column(public_ip);size(128)"`
	PrivateIp      string    `json:"private_ip" orm:"column(private_ip);size(128)"`
	TaskId         string    `orm:"column(task_id);size(128)"`
	State          string    `json:"state" orm:"column(state);size(128)"`
	ErrMsg         string    `json:"err_msg" orm:"column(err_msg);size(128)"`
	ActiveTime     time.Time `orm:"column(active_time);type(datetime);null"`
	TimeModel
}

func (p *PodInfo) TableIndex() [][]string {
	return [][]string{
		{"HostName"},
		{"TaskId"},
		{"FleetId"},
		{"ScalingGroupId"},
		{"State"},
	}
}

type podInfoTable struct{}

var podTable = podInfoTable{}

func PodInfoTable() *podInfoTable {
	return &podTable
}

func (p *podInfoTable) Insert(info *PodInfo) error {
	_, err := ormer.Insert(info)
	return err
}

func (p *podInfoTable) Update(info *PodInfo, cols ...string) error {
	_, err := ormer.Update(info, cols...)
	return err
}

func (p *podInfoTable) Get(f Filters) (*PodInfo, error) {
	var pod PodInfo
	err := f.Filter(tableNamePodInfo).One(&pod)
	return &pod, err
}

func (p *podInfoTable) GetPodByHostName(hostName string) (*PodInfo, error) {
	var pod PodInfo
	err := ormer.QueryTable(tableNamePodInfo).Filter(fieldNameHostName, hostName).One(&pod)
	return &pod, err
}

func (p *podInfoTable) GetFailJob(taskId string) (int, []PodInfo, error) {
	var failPods = []PodInfo{}
	count, err := ormer.QueryTable(tableNamePodInfo).Filter(fieldNameTaskId, taskId).Filter(fieldNameState, PodStatePending).All(&failPods)
	return int(count), failPods, err
}

func (p *podInfoTable) DeletePodInfoByGroupId(groupId string) error {
	_, err := ormer.QueryTable(tableNamePodInfo).Filter(fieldNameScalingGroupId, groupId).Update(orm.Params{
		fieldNameState:     PodStateDeleted,
		fieldNameIsDeleted: deletedFlag,
		fieldNameDeleteAt:  time.Now().UTC(),
	})
	if err != nil {
		return err
	}
	return nil
}

func (p *podInfoTable) TxUpdateState(hostName, newState string) error {
	return ormer.DoTx(func(ctx context.Context, txOrm orm.TxOrmer) error {
		podInfo := &PodInfo{}
		err := txOrm.QueryTable(tableNamePodInfo).
			Filter(fieldNameHostName, hostName).
			One(podInfo)
		if err != nil {
			return err
		}
		num, err := txOrm.QueryTable(tableNamePodInfo).
			Filter(fieldNameHostName, hostName).
			Filter(fieldNameState, podInfo.State).Update(
			orm.Params{
				fieldNameState:    newState,
				fieldNameUpdateAt: time.Now().UTC(),
			})
		if err != nil {
			return err
		}
		if num == 0 {
			logger.R.Error("cannot transfer pod [%s] state from %s to %s, because state has been changed",
				hostName, podInfo.State, newState)
			return fmt.Errorf("cannot transfer pod [%s] state from %s to %s, because state has been changed",
				hostName, podInfo.State, newState)
		}
		logger.R.Info("transfer pod [%s] from state %s to new state %s success", hostName, podInfo.State, newState)
		return nil
	})
}

func (p *podInfoTable) UpdatePodActiveTime(hostName string) error {
	_, err := ormer.QueryTable(tableNamePodInfo).
		Filter(fieldNameHostName, hostName).
		Update(orm.Params{
			fieldNameActiveTime: time.Now().UTC(),
		})
	return err
}

func (p *podInfoTable) GetPodsByFilter(f Filters) ([]PodInfo, error) {
	var Pods []PodInfo
	_, err := f.Filter(tableNamePodInfo).All(&Pods)
	return Pods, err
}

// 软删除
func (p *podInfoTable) DeletePod(hostName string) error {
	_, err := ormer.QueryTable(tableNamePodInfo).Filter(fieldNameHostName, hostName).Update(orm.Params{
		fieldNameState:     PodStateDeleted,
		fieldNameIsDeleted: deletedFlag,
		fieldNameDeleteAt:  time.Now().UTC(),
	})
	return err
}

// 硬删除
func (p *podInfoTable) Delete(hostName string) error {
	_, err := ormer.QueryTable(tableNamePodInfo).Filter(fieldNameHostName, hostName).Delete()
	return err
}

func (p *podInfoTable) ListPod(groupId, state string, limit, offset int) (int, int, []PodInfo, error) {
	var podList []PodInfo
	qs := ormer.QueryTable(tableNamePodInfo)
	condition := orm.NewCondition()
	if state != "" {
		condition = condition.And(fieldNameState, state)
	} else {
		condition = condition.And(fieldNameStateIn, PodStateRunning, PodStatePending, PodStateActive)
	}
	condition = condition.And(fieldNameScalingGroupId, groupId)
	count, err := qs.SetCond(condition).Limit(limit).Offset(offset * limit).OrderBy(DefaultSort).All(&podList)
	if err != nil {
		return 0, 0, nil, err
	}
	total, err := qs.SetCond(condition).Count()
	if err != nil {
		return 0, 0, nil, err
	}
	return int(count), int(total), podList, nil
}

func (p *podInfoTable) GetAllPodNames(groupId string, states []string) ([]string, error) {
	var podList []PodInfo
	_, err := ormer.QueryTable(tableNamePodInfo).Filter(fieldNameScalingGroupId, groupId).
		Filter(fieldNameStateIn, states).
		Filter(fieldNameIsDeleted, notDeletedFlag).OrderBy("-state", "-create_at").All(&podList)
	if err != nil {
		return nil, err
	}
	var podNames []string
	for _, pod := range podList {
		podNames = append(podNames, pod.HostName)
	}
	return podNames, nil
}

// 用deleting标记正在被缩容，表示已经移出伸缩组
func (p *podInfoTable) RemoveInstance(pods []string) error {
	_, err := ormer.QueryTable(tableNamePodInfo).
		Filter(fieldNameHostNameIn, pods).
		Update(orm.Params{
			fieldNameState: PodStateDeleting,
		})

	return err
}

func (p *podInfoTable) CalculatePodNumOfGroup(groupId string) (int, error) {
	var count int64
	count, err := ormer.QueryTable(tableNamePodInfo).
		Filter(fieldNameScalingGroupId, groupId).
		Filter(fieldNameStateIn, PodStateActive, PodStateRunning).Count()
	return int(count), err

}

// 事务更新当前实例数
func (p *podInfoTable) TxUpdateCurNum(groupId string, curNum int) error {
	return ormer.DoTx(func(ctx context.Context, txOrm orm.TxOrmer) error {
		// 当伸缩组稳定时，update current number的affect row为0，此时updated_at不会更新
		_, err := txOrm.QueryTable(tableNameScalingGroup).
			Filter(fieldNameId, groupId).
			Filter(fieldNameIsDeleted, notDeletedFlag).
			Update(orm.Params{
				fieldNameCurNum: curNum,
			})
		if err != nil {
			return err
		}
		return nil
	})

}

func IsPodExit(hostName string) bool {
	return ormer.QueryTable(tableNamePodInfo).
		Filter(fieldNameHostName, hostName).
		Filter(fieldNameIsDeleted, notDeletedFlag).
		Exist()
}

func GetPodInfoByHostName(hostName string) (*PodInfo, error) {
	pod := PodInfo{
		HostName: hostName,
	}
	err := ormer.Read(&pod)

	if err != nil {
		return nil, err
	}
	return &pod, err
}

func CleanUpPodInfoTable(dayBefore int, log *logger.FMLogger) error {
	logger.R.Info("start to clean pod info table ...")
	timeLimit := time.Now().UTC().AddDate(0, 0, -dayBefore).Format(common.TimeLayout)
	condition := orm.NewCondition()
	condition = condition.And(fieldNameUpdateAtLt, timeLimit)
	var podList = &[]PodInfo{}
	count, err := ormer.QueryTable(tableNamePodInfo).SetCond(condition).All(podList)
	if err != nil {
		log.Error("get ecs info table before %d days ago data with error:%+v", dayBefore, err)
		return err
	}

	wg := sync.WaitGroup{}
	wg.Add(int(count))
	for index, pod := range *podList {
		if index > 0 && index%common.InstanceCountToSleep == 0 {
			time.Sleep(common.InstanceCountToSleepSeconds)
		}
		go func(HostName string) {
			defer wg.Done()
			if err := PodInfoTable().Delete(HostName); err != nil {
				log.Error("delete pod [%s] error:%+v", HostName, err)
			}
			log.Info("delete pod [%s] success", HostName)
		}(pod.HostName)
	}
	wg.Wait()
	return nil
}

// 识别超时未上报的实例
func VerifyZombiePod(log *logger.FMLogger, workNodeWG *sync.WaitGroup) error {
	defer workNodeWG.Done()
	var zombieInstance []PodInfo
	q := ormer.QueryTable(tableNamePodInfo).
		Filter(fieldNameStateIn, PodStateActive, PodStateRunning).
		Filter(fieldNameUpdateAtLt, time.Now().UTC().Add(-common.DefaultVMTimeOutDuration*time.Minute).Format(common.TimeLayout))
	zombieCount, err := q.All(&zombieInstance)
	if err != nil {
		return err
	}
	log.Info("finish verify %d zombie Instance, start to transfer state", zombieCount)
	if zombieCount == 0 {
		return nil
	}
	wg := sync.WaitGroup{}
	wg.Add(int(zombieCount))
	for idx, _ := range zombieInstance {
		if idx > 0 && idx%common.InstanceCountToSleep == 0 {
			time.Sleep(common.InstanceCountToSleepSeconds * time.Second)
		}
		go func(pod PodInfo) {
			defer wg.Done()
			if err := PodInfoTable().TxUpdateState(pod.HostName, PodStateError); err != nil {
				log.Error("update pod %s state to Error fail with error:%+v", pod.HostName, err)
				return
			}
			log.Info("instance id %s is zombie pod, last update at %s", pod.HostName, pod.UpdateAt)
		}(zombieInstance[idx])
	}
	wg.Wait()

	return nil
}
